<?php

declare(strict_types=1);

namespace App\Amqp\Consumer;

use Hyperf\Amqp\Annotation\Consumer;
use Hyperf\Amqp\Message\ConsumerMessage;
use Hyperf\Amqp\Result;
use PhpAmqpLib\Message\AMQPMessage;
use Hyperf\Amqp\Message\Type;
use Service\Code\Normal;
use Service\AppService\MonitorService;
use Hyperf\Amqp\Producer;
use App\Amqp\Producer\ChannelUserTagNotNullMessageProducer;
use App\Amqp\Producer\ChannelSelfBbcMessageProducer;
use App\Amqp\Producer\ChannelOuterOpenUserMessageProducer;
use App\Amqp\Producer\ChannelNextLevelYunCangMessageProducer;
use Service\ConsumerService\BbcChannelCommodityConsumerService;

/**
 * qos (prefetch_size,prefetch_count,global)
 * @Consumer(exchange="COMMODITY_BBC_MICRO_SERVICE_MESSAGE", routingKey="COMMODITY_BBC_MICRO_SERVICE_MESSAGE", queue="COMMODITY_BBC_MICRO_SERVICE_MESSAGE", nums=1)
 */
class ChannelCommodityMessageConsumer extends ConsumerMessage
{
    protected $type = Type::DIRECT;//直连模式
    protected $qos  = [
        'prefetch_count'    => 10
    ];
    public function consumeMessage($data, AMQPMessage $message): string
    {
        try {
            if( empty($data['MessageType']) ) {
                echo "current MessageType is Missing\r\n";
                sleep(3);//休眠3秒之后，重回队列
                return Result::REQUEUE;
            }
            
            if( empty($data['DataList']) ) {
                echo "current DataList is Missing\r\n";
                sleep(3);//休眠3秒之后，重回队列
                return Result::REQUEUE;
            }
            
            //开始消费商品变动
            $result = BbcChannelCommodityConsumerService::instance()->consumer($data['MessageType'],$data['DataList']);
            if( $result['code'] == Normal::SUC ) {
                return Result::ACK;
            } else {
                $contents = json_encode($result,JSON_UNESCAPED_UNICODE);
                MonitorService::instance()->sendMsg($contents);
                sleep(3);//休眠3秒之后，重回队列
                return Result::REQUEUE;
            }
            
        } catch ( \Throwable $t ) {
            $error = [
                'error'         => $t->getMessage(),
                'errorFile'     => $t->getFile(),
                'errorLine'     => $t->getLine(),
            ];
            
            $contents = json_encode($error,JSON_UNESCAPED_UNICODE);
            MonitorService::instance()->sendMsg($contents);
            sleep(10);//休眠10秒之后，重回队列
            return Result::REQUEUE;
        }
    }
    
    /**
     * 设置是否自启动
     */
    public function isEnable() : bool
    {
        $isDisAable = env('AMQP_SHOP_COMMODITY_CHANNEL_MESSAGE_CONSUMER_DISABLE');
        return $isDisAable ? false : true ;
    }
    
    
    /**
     * 转移队列数据
     */
    public function pushChannelNextLevelUserMesage($data)
    {
        try
        {
            $message = new ChannelNextLevelYunCangMessageProducer($data);
            $producer = container()->get(Producer::class);
            $result = $producer->produce($message);
            if($result) {
                return Normal::returnCode(Normal::SUC,"发布成功");
            }
            return Normal::returnCode(Normal::FAIL,"发布消息队列失败");
        }
        catch ( \Throwable $th )
        {
            return Normal::returnCode(Normal::FAIL,"发布消息队列失败【{$th->getMessage()}】");
        }
    }
    
    /**
     * 转移队列数据
     */
    public function pushChannelOuterOpenUserMessage($data)
    {
        try
        {
            $message = new ChannelOuterOpenUserMessageProducer($data);
            $producer = container()->get(Producer::class);
            $result = $producer->produce($message);
            if($result) {
                return Normal::returnCode(Normal::SUC,"发布成功");
            }
            return Normal::returnCode(Normal::FAIL,"发布消息队列失败");
        }
        catch ( \Throwable $th )
        {
            return Normal::returnCode(Normal::FAIL,"发布消息队列失败【{$th->getMessage()}】");
        }
    }
    
    /**
     * 转移队列数据
     */
    public function pushChannelSelfBbcMessage($data)
    {
        try
        {
            $message = new ChannelSelfBbcMessageProducer($data);
            $producer = container()->get(Producer::class);
            $result = $producer->produce($message);
            if($result) {
                return Normal::returnCode(Normal::SUC,"发布成功");
            }
            return Normal::returnCode(Normal::FAIL,"发布消息队列失败");
        }
        catch ( \Throwable $th )
        {
            return Normal::returnCode(Normal::FAIL,"发布消息队列失败【{$th->getMessage()}】");
        }
    }
    
    /**
     * 转移队列数据
     */
    public function pushChannelUserTagNotNullMessage($data)
    {
        try
        {
            $message = new ChannelUserTagNotNullMessageProducer($data);
            $producer = container()->get(Producer::class);
            $result = $producer->produce($message);
            if($result) {
                return Normal::returnCode(Normal::SUC,"发布成功");
            }
            return Normal::returnCode(Normal::FAIL,"发布消息队列失败");
        }
        catch ( \Throwable $th )
        {
            return Normal::returnCode(Normal::FAIL,"发布消息队列失败【{$th->getMessage()}】");
        }
    }
}